package com.cskaoyan.mr.topn;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.TreeSet;

/**
 * @ClassName TopNReducer
 * @Description: TODO
 * @Author 远志 zhangsong@cskaoyan.onaliyun.com
 * @Date 2023/2/13 14:42
 * @Version V1.0
 **/
public class TopNReducer extends Reducer<IntWritable, NullWritable, IntWritable, NullWritable> {

    private TreeSet<Integer> treeSet = new TreeSet<>();

    @Override
    protected void reduce(IntWritable key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        //只需要确保只有唯一的一个reducer，那么所有数据均会到这个方法中来
        //实际上，默认情况下reducer就是一个
        treeSet.add(key.get());
        if(treeSet.size() > TopN.N){
            treeSet.pollFirst();
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        //treeSet.descendingSet()
        for (Integer integer : treeSet) {
            context.write(new IntWritable(integer), NullWritable.get());
        }
    }
}
